//package com.bs.kafka.config;
//
//import lombok.extern.log4j.Log4j2;
//import org.apache.kafka.clients.consumer.Consumer;
//import org.apache.kafka.clients.consumer.ConsumerRecord;
//import org.apache.kafka.clients.consumer.OffsetAndMetadata;
//import org.apache.kafka.clients.consumer.OffsetCommitCallback;
//import org.apache.kafka.common.TopicPartition;
//import org.springframework.context.annotation.Configuration;
//import org.springframework.kafka.annotation.KafkaListener;
//import org.springframework.kafka.core.KafkaTemplate;
//import org.springframework.kafka.support.Acknowledgment;
//
//import javax.annotation.Resource;
//import java.util.List;
//import java.util.Map;
//
///**
// * @author bingshao
// * @date 2022/10/10
// **/
//@Configuration
//@Log4j2
//public class ConsumerConfig {
//
//    @Resource
//    private KafkaTemplate<String, Object> kafkaTemplate;
//
//    //    @KafkaListener(topics = {"test1"})
////    public void consumer1(List<ConsumerRecord<?, ?>> consumerRecordList, Consumer consumer) {
////        consumerRecordList.forEach(record -> {
////            System.out.println("简单消费1===Topic：" + record.topic() + "**分区" + record.partition() + "**值内容" + record.value() + "**位移" + record.offset());
////            int a =  100/0;
////        });
////    }
//
//    @KafkaListener(topics = {"test1"}, errorHandler = "kafkaRetryHandler")
//    public void consumer1(ConsumerRecord<?, ?> record, Consumer consumer) {
//        // 失败测试
//        int a = 100 / 0;
//        log.info("简单消费1===Topic：" + record.topic() + "**分区" + record.partition() + "**值内容" + record.value() + "**位移" + record.offset());
//    }
//
//    @KafkaListener(topics = {"test1.DLT"})
//    public void consumer2(ConsumerRecord<?, ?> record, Consumer consumer) {
//        try {
//            log.info("死性消费===Topic：" + record.topic() + "**分区" + record.partition() + "**值内容" + record.value() + "**位移" + record.offset());
//            consumer.commitAsync();
//        } catch (Exception e) {
//
//        }
//
//    }
//
////    @KafkaListener(topics = {"flink-test"})
////    public void consumer3(ConsumerRecord<?, ?> record, Consumer consumer) {
////        try {
////            log.info("flink-test===Topic：" + record.topic() + "**分区" + record.partition() + "**值内容" + record.value() + "**位移" + record.offset());
////            consumer.commitAsync();
////        } catch (Exception e) {
////
////        }
////
////    }
//
//    @KafkaListener(topics = {"flink-send"})
//    public void consumer4(ConsumerRecord<?, ?> record, Consumer consumer) {
//        try {
//            log.info("flink-send===Topic：" + record.topic() + "**分区" + record.partition() + "**值内容" + record.value() + "**位移" + record.offset());
//            consumer.commitAsync();
//        } catch (Exception e) {
//
//        }
//
//    }
//
//}
